package com.htzz.collection.config;

import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.annotation.PostConstruct;

import cn.hutool.core.util.StrUtil;
import com.alibaba.druid.support.json.JSONUtils;
import com.htzz.collection.service.CollectionService;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Configuration;
import com.alibaba.fastjson.JSONObject;

import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.PropertySource;

/**
 * mqtt 消费者类
 * @author xudezhi
 *
 */
@Slf4j
@Configuration
@ConditionalOnProperty(prefix = "mqtt",name = "use", havingValue = "true", matchIfMissing = false)
public class MqttReceiveConfig implements MqttCallbackExtended {

	ExecutorService pool = Executors.newFixedThreadPool(1000);

	MqttClient client;
	
	MqttConnectOptions options;

	@Value("${mqtt.broker.serverUri}")
	private String MqttUrl;

	@Value("${mqtt.broker.username}")
	private String userName;
	@Value("${mqtt.broker.password}")
	private String password;

	@Value("${mqtt.topic}")
	protected String topic;

	@Value("${mqtt.qos}")
	protected int qos;

	@Value("${mqtt.broker.cleansession}")
	private boolean cleansession;

	@Value("${mqtt.broker.automaticreconnect}")
	private boolean automaticreconnect;

	@Autowired
	private CollectionService collectionService;

	private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
	/**
	 * 初始化mqtt消费者
	 */
	@PostConstruct
	public void init() {
		try {
			client = new MqttClient(MqttUrl, UUID.randomUUID().toString(), new MemoryPersistence());
			log.info("初始化mqtt客户端"+client);
			
			log.info("配置mqtt客户端配置信息");
	    	options = new MqttConnectOptions();
	    	options.setUserName(userName);
	    	options.setPassword(password.toCharArray());
			options.setCleanSession(cleansession);
			options.setConnectionTimeout(30);
			options.setKeepAliveInterval(15);
			options.setAutomaticReconnect(automaticreconnect);
			options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_DEFAULT);
			
			client.setCallback(this);
			client.connect(options);
			client.subscribe(topic, qos);
			log.info("初始化mqtt客户端！");
		} catch (MqttSecurityException e) {
			log.info("初始化mqtt客户端异常：MqttSecurityException！\n"+e.getLocalizedMessage());
			e.printStackTrace();
		} catch (MqttException e) {
			log.info("初始化mqtt客户端异常：MqttException！\n"+e.getLocalizedMessage());
			e.printStackTrace();
		}

	}
	
	

	/**
	 * 断开链接，重连操作
	 */
	public void connectionLost(Throwable cause) {
		log.info("[MQTT] 连接断开，1S之后尝试重连...");
		while (true) {
			try {
				if (!client.isConnected()) {
					log.info("重连mqtt客户端");
					client.setCallback(this);
					client.connect(options);
				}
				if (client.isConnected()) {
					log.info("mqtt客户端已重新链接");
					break;
				}
			} catch (MqttException e) {
				log.info("mqtt重连异常");
			}
		}
	}
	

	/**
	 * 连接成功
	 */
	@Override
	public void connectComplete(boolean reconnect, String serverURI) {
		log.info("开始订阅.............");
		try {
			client.subscribe(topic, qos);
			log.info("订阅主题："+topic+",qos:"+qos);
		} catch (MqttException e) {
			log.info("订阅主题："+topic+",qos:"+qos+"异常："+e.getLocalizedMessage());
			e.printStackTrace();
		}
		log.info("已经订阅.........");
	}

	/**
	 * 收到新的消息推送
	 * 示例：
	 * {
	 * 	"devId":"sn19990111",
	 * 	"t": 1832893283813,
	 * 	"data":{
	 * 		"speed": "3232",
	 * 		"cc":"ddd"
	 *        }
	 * }
	 */
	public void messageArrived(String topic, MqttMessage message) throws Exception {
		log.info("Consumer接收到的数据\n topic:" + topic + "\n message:" + message.toString());
		
		//消息内容
		String msg = message.toString().trim();
		try {
			pool.submit(new Runnable() {
				@Override
				public void run() {

					log.info("检查消息格式");
					//内容转换成JSON
					JSONObject json = JSONObject.parseObject(msg);
					log.info("----------------------数据格式正确---------------------");

					String devId = null;
					//设备ID判断
					try {
						devId = (String) json.get("devId");
						log.info("获取到的设备id为：" + devId);
						if(StrUtil.isEmpty(devId)){
							log.info("设备ID为空");
							return;
						}
					} catch (Exception e) {
						log.info("获取设备ID异常");
						return;
					}
					log.info("设备信息: devId:"+ devId);
					/**
					 * 组装存储所需数据
					 */
					String nowTime = sdf.format(new Date());
					Object t = json.get("t");
					String tTime="";
					if (null == t) {
						tTime=nowTime;
					} else {
						Calendar inst = Calendar.getInstance();
						inst.setTimeInMillis(Long.parseLong(t.toString()));
						tTime= sdf.format(inst.getTime());
					}
					log.info("时间: t:"+ tTime);

					//解析所有消息
					JSONObject dataObj = (JSONObject) json.get("data");

					List<Map<String, Object>> datas=new ArrayList<>();
					for(String key :dataObj.keySet()){ //处理其他属性
						if(!key.equals("t")) {
							HashMap<String, Object> mqttArg = new HashMap<String, Object>();
							mqttArg.put("createTime", nowTime);
							mqttArg.put("devId", devId);
							mqttArg.put("t", tTime);
							mqttArg.put("k",key);
							mqttArg.put("v",dataObj.get(key));
							log.info("调用存储采集信息："+JSONObject.toJSONString(mqttArg));
//							collectionService.saveData(finalDevId,mqttArg);
							datas.add(mqttArg);
						}
					}
					collectionService.saveDatas(devId,datas);
				}
			});
		} catch (Exception e) {
			log.info("----------------------数据格式不正确---------------------");
			log.info("格式错误信息："+e.getLocalizedMessage());
		}
	}


	/**
	 * 消息处理结束
	 */
	public void deliveryComplete(IMqttDeliveryToken token) {
		log.info("消息处理结束时调用");
	}
}
